package com.rabbitmq.learn.confirm.unsync;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.learn.common.ConnectionUtils;
import com.rabbitmq.learn.common.RabbitMqConstant;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * Created by Lxk on 2019/8/17.
 */
public class Sender {

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection con = ConnectionUtils.getConnection();
        Channel channel = con.createChannel();
        channel.queueDeclare(RabbitMqConstant.ConfirmUnsync.QUEUE_CONFIRM_UNSYNC,false,false,false,null);

        channel.confirmSelect();

        final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        channel.addConfirmListener(new ConfirmListener() {

            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    System.out.println("handle multiple ack");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{
                    System.out.println("handle single ack");
                    confirmSet.remove(deliveryTag);
                }
            }

            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    System.out.println("handle multiple Nack");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{
                    System.out.println("handle single Nack");
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        String prefiex = "unsync msg";
        int count = 0;
        while(true){
            String msg = prefiex + count;
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("",RabbitMqConstant.ConfirmUnsync.QUEUE_CONFIRM_UNSYNC,null,msg.getBytes());
            confirmSet.add(seqNo);
            count++;
        }

    }

}
